package com.dycong.common.Lamda;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * @author dycong
 * @Link https://mp.weixin.qq.com/s/W_cm46lAmQFgLP0gw0UcDw
 * @date 2019/6/3 9:22
 */
public class CompletionOrderSpliterator<T> implements Spliterator<T> {


    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10, t -> {
            Thread thread = new Thread(t);
            thread.setDaemon(true);
            return thread;
        });

        List<CompletableFuture<Integer>> futures = Stream
                .iterate(0, i -> i + 1)
                .limit(100)
                .map(i -> CompletableFuture.supplyAsync(
                        withRandomDelay(i), executorService))
                .collect(Collectors.toList());

        parallelCompletionOrder(futures).parallel()
                .forEach(o -> {
                    System.out.println(o + "  " + Thread.currentThread());
                });
    }

    public static <T> Stream<T> completionOrder(Collection<CompletableFuture<T>> futures) {
        return StreamSupport.stream(
                new CompletionOrderSpliterator<>(futures), false);
    }

    public static <T> Stream<T> parallelCompletionOrder(Collection<CompletableFuture<T>> futures) {
        return StreamSupport.stream(
                new CompletionOrderSpliterator<>(futures), true);
    }

    public static <T> Stream<T> originalOrder(Collection<CompletableFuture<T>> futures) {
        return futures.stream().map(CompletableFuture::join);
    }

    public static <T> Stream<T> parallelOriginalOrder(Collection<CompletableFuture<T>> futures) {
        return futures.parallelStream().map(CompletableFuture::join);
    }

    private static Supplier<Integer> withRandomDelay(Integer i) {
        return () -> {
            try {
                Thread.sleep(ThreadLocalRandom.current()
                        .nextInt(1000));
                System.out.println(Thread.currentThread());
            } catch (InterruptedException e) {
                // ignore shamelessly, don't do this on production
            }
            return i;
        };
    }

    private final Map<Integer, CompletableFuture<Map.Entry<Integer, T>>> indexedFutures;

    CompletionOrderSpliterator(Collection<CompletableFuture<T>> futures) {
        indexedFutures = toIndexedFutures(futures);
    }

    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        if (!indexedFutures.isEmpty()) {
            action.accept(nextCompleted());
            return true;
        } else {
            return false;
        }
    }

    private T nextCompleted() {
        return CompletableFuture.anyOf(indexedFutures.values().toArray(new CompletableFuture[0]))
                .thenApply(result -> ((Map.Entry<Integer, T>) result))
                .thenCompose(result -> indexedFutures.remove(result.getKey()))
                .thenApply(Map.Entry::getValue)
                .join();
    }

    // TODO: 2019/6/3 不切分就不会有ForkJoinPool线程并发执行  Thread[ForkJoinPool.commonPool-worker-1,5,main]
    @Override
    public Spliterator<T> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return indexedFutures.size();
    }

    @Override
    public int characteristics() {
        return SIZED & IMMUTABLE & NONNULL;
    }

    private static <T> Map<Integer, CompletableFuture<Map.Entry<Integer, T>>> toIndexedFutures(Collection<CompletableFuture<T>> futures) {
        Map<Integer, CompletableFuture<Map.Entry<Integer, T>>> map = new HashMap<>(futures.size(), 1);

        int counter = 0;
        for (CompletableFuture<T> f : futures) {
            int index = counter++;
            map.put(index, f.thenApply(value -> new AbstractMap.SimpleEntry<>(index, value)));
        }
        return map;
    }
}
